feat: versioned JobManager for reading/writing jobs across protocol versions#9434
feat: versioned JobManager for reading/writing jobs across protocol versions#9434koenvanderveen wants to merge 34 commits into
Conversation
…es; rename downgrade to protocol version
…ame write-target schema helper
Loop the p2p compatibility test over every released on-disk fixture (keyed on release, e.g. syft_job-0.1.38-protocol0_syftbox) instead of a single protocol, asserting jobs load/upgrade and that scan/approve write back byte-exact in each release's layout/format. Also check job reprs render without error. Add scripts/generate_release_fixture.py to snapshot the current release's SyftBox tree (run on every release) and move export_release_artifact.py alongside it in scripts/. Document both in CLAUDE.md and README. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
datasite_owner_email, submitted_by and name now read from the JobRef (parsed from the folder layout, governed by syft permissions) instead of the DS-writable config.yaml, so a data scientist can't spoof them. Make ref required and drop the metadata-based fallback; current_user_email stays a stored arg from the client config. Add a test proving a tampered config.yaml name is ignored in favor of the path. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…efactor Pass the path-derived JobRef and drop the removed datasite_owner_email kwarg. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
pjwerneck
left a comment
pjwerneck
left a comment
There was a problem hiding this comment.
The comments mark some minor issues found in the code.
The main high-level design issue would be having a separate ProtocolCodec class, to avoid bundling all protocol rules in the _load_upgraded and _write_in_target_version methods, as suggested in the comments.
| self.service = MigrationService(registry=registry) | ||
| # peer email -> job ProtocolSchema; filled in by syft-client later. | ||
| # Peers without an entry are assumed to run the current protocol. | ||
| self.peer_schemas: dict[str, ProtocolSchema] = peer_schemas or {} |
There was a problem hiding this comment.
While peer_schemas is being assigned here and the versioning machinery treats it correctly, it's apparently not being sent by the syft-client, since JobClient is initialized using .from_config and there's no peer_schemas being passed anywhere.
The test cases are correct and pass peer_schemas, but the actual code doesn't.
| # -- model IO --------------------------------------------------------------- | ||
| def read_submission(self, ref: JobRef) -> JobSubmissionMetadata: | ||
| """Load a submission's config.yaml, upgraded to the latest version.""" | ||
| extra = {"submitted_by": ref.ds_email, "datasite_email": ref.datasite_email} |
There was a problem hiding this comment.
Shouldn't ref.job_name be added to extra here too, since it comes from the path like submitted_by and datasite_email?
| ] | ||
| if not matches: | ||
| raise FileNotFoundError(f"Job '{job_name}' not found") | ||
| if len(matches) > 1: |
There was a problem hiding this comment.
This will raise an error with jobs with the same name across protocol versions. Is that intended?
| except Exception: | ||
| continue | ||
| ) | ||
| except Exception: |
There was a problem hiding this comment.
Blanket exception catches on a long block like this are an anti-pattern. It should catch specific exceptions, or at least log the error with the job name and path.
| self, | ||
| ds_email: str, | ||
| job_name: str, | ||
| protocol_version: str = JOB_PROTOCOL_VERSION, |
There was a problem hiding this comment.
Since receive_job is only called by scan_inbox, which already determines the correct protocol_version on load, it should be required parameter, not default to the constant.
| # Update state to RUNNING | ||
| state_file = review_dir / "state.yaml" | ||
| state = JobState.load(state_file) | ||
| state = self.manager.read_state(ref) |
There was a problem hiding this comment.
JobManager.read_state can return None if the state.yaml no longer exists. There should be a check here to prevent an exception on the state.status on the next line.
| return path | ||
|
|
||
| # -- internals ----------------------------------------------------------------- | ||
| def _load_upgraded( |
There was a problem hiding this comment.
Both _load_upgraded and _write_in_target_version work for only two versions, but are somewhat fragile when it comes to handling more future versions, and specially once we need to upgrade or downgrade jobs throughout multiple versions.
Instead of conditionals I would introduce a ProtocolCodec class that handles the serialization and loading for each version, and JobManager delegates to the appropriate class, keeping all migration specific code in the codec class. Once v2 is introduced, it's just a new class, instead of more conditionals.
| protocol_version: str # "0" (no path segment) or "1"+ (v<n> segment) | ||
|
|
||
|
|
||
| class JobManager: |
There was a problem hiding this comment.
JobManager is a somewhat vague name that blurs the line with JobClient. Although the two have clearly distinct responsibilities, the suffix is not as clear. Maybe JobRepository or JobStorage.
| return None | ||
| return self._load_upgraded(path, "JobState", {}) | ||
|
|
||
| def write_submission(self, ref: JobRef, metadata: JobSubmissionMetadata) -> Path: |
There was a problem hiding this comment.
Both write_submission and write_state could be renamed to make the asymmetry more explicit, like in the docstring. Like write_submission_for_do and write_state_for_ds.
| if review_state.status == JobStatus.FAILED: | ||
| self.state.mark_notified(metadata.name, "failed") | ||
| count += 1 | ||
| self.state.mark_notified(metadata.name, "new") |
There was a problem hiding this comment.
metadata.name was replaced with ref.job_name in most places, but still remains in others like here.
2 participants
JobManager(nested inJobClient) owns allJobState/JobSubmissionMetadatafilesystem IO: reads upgrade objects in memory to the latest registered version, writes downgrade them to the version/layout the reading peer understands.JobClient,JobInfo,SyftJobRunner, the syft-bg job monitor, and the enclave client all route through it.inbox/<ds_email>/v1/<job_name>/(same forreview/). Protocol 0 = last release (0.1.38): no version segment, nocanonical_name/versionfields in yaml; protocol-0 jobs are read, listed, and written back byte-compatibly with 0.1.38.JobClientaccepts optionalpeer_schemas(peer email → jobProtocolSchema); the version spoken to a peer is the min of ours and theirs, so an older peer receives the older layout/format. Unknown peers default to the current protocol.MigrationRegistrygains aprotocol_version; released schemas are frozen asReleasedProtocol/ReleasedPackageProtocolInfoartifacts (save/load, registered on import) with drift detection that fails if a released object schema changes without a new version. A hardcoded 0.1.38 artifact ships insyft_job/migrations/history/as if that release had emitted it.JobInfoidentity from the path, notconfig.yaml:datasite_owner_email,submitted_by, andnamenow come from the path-derivedJobRef(governed by syft permissions), never from the DS-writableconfig.yaml— closing a spoofing vector where a DS could claim a fake identity.refis now required;current_user_emailstays sourced from the client config.packages/syft-job/scripts/):export_release_artifact.pywrites the JSON schema artifacts, and the newgenerate_release_fixture.pysnapshots a full SyftBox tree of exactly how the release serializes jobs to disk. Both are run on every release and documented inCLAUDE.md/README.md.syft_job-<version>-protocol<p>_syftbox(0.1.38/protocol 0 hand-authored, 0.1.39/protocol 1 generated).test_older_protocol_compatibility.pyloops over all of them — keyed on release, not just protocol, since a release can change how it writes bytes without bumping the protocol — asserting load/upgrade plus byte-exact scan/approve write-back per protocol.Test plan: syft-migration + syft-job migration/p2p suites (list/upgrade, scan, approve write-back byte-match, mixed listing, reserved
v<n>job names, per-protocol peer submit, perms on thev1subfolder, repr smoke test, and an identity-from-path spoofing test) plus roottests/unit; all suites green, pre-commit clean, history JSON verified in the wheel.